/**
* MagicProtocol.java 2020/5/16 11:27 上午
* Copyright ©2020 www.bmsoft.com.cn All rights reserved.
* PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
*/
package protocol.magic;

import common.constants.CommonConstants;
import common.exception.MagicRpcException;
import common.exception.RemotingException;
import common.factory.MagicFactory;
import common.model.Exporter;
import common.model.Invocation;
import common.model.Invoker;
import common.model.MagicUrl;
import common.model.MagicUrlBuilder;
import common.model.Result;
import common.model.RpcContext;
import common.model.RpcInvocation;
import common.util.CollectionUtils;
import common.util.ConcurrentHashSet;
import common.util.ConfigUtils;
import common.util.NetUtils;
import common.util.StringUtils;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
import protocol.AbstractProtocol;
import protocol.LazyConnectExchangeClient;
import protocol.ReferenceCountExchangeClient;
import remoting.channel.Channel;
import remoting.exchange.ExchangeChannel;
import remoting.exchange.ExchangeClient;
import remoting.exchange.ExchangeHandler;
import remoting.exchange.ExchangeServer;
import remoting.exchange.Exchangers;
import remoting.exchange.support.ExchangeHandlerAdapter;

import static common.constants.CommonConstants.GROUP_KEY;
import static common.constants.CommonConstants.INTERFACE_KEY;
import static common.constants.CommonConstants.PATH_KEY;
import static common.constants.CommonConstants.VERSION_KEY;
import static common.constants.CommonConstants.LAZY_CONNECT_KEY;
import static protocol.magic.Constants.ON_CONNECT_KEY;
import static protocol.magic.Constants.ON_DISCONNECT_KEY;
import static common.constants.CommonConstants.DEFAULT_HEARTBEAT;
import static common.constants.CommonConstants.HEARTBEAT_KEY;
import static common.constants.CommonConstants.CHANNEL_READONLYEVENT_SENT_KEY;
import static common.constants.CommonConstants.CLIENT_KEY;
import static common.constants.CommonConstants.DEFAULT_REMOTING_CLIENT;
import static common.constants.CommonConstants.SERVER_KEY;
import static common.constants.CommonConstants.DEFAULT_REMOTING_SERVER;
import static protocol.magic.Constants.CALLBACK_SERVICE_KEY;
import static common.constants.CommonConstants.CONNECTIONS_KEY;
import static common.constants.CommonConstants.DEFAULT_STUB_EVENT;
import static protocol.magic.Constants.IS_CALLBACK_SERVICE;
import static common.constants.CommonConstants.IS_SERVER_KEY;
import static common.constants.CommonConstants.STUB_EVENT_KEY;
import static common.constants.CommonConstants.STUB_EVENT_METHODS_KEY;
import static protocol.magic.Constants.SHARE_CONNECTIONS_KEY;
import static protocol.magic.Constants.DEFAULT_SHARE_CONNECTIONS;

/**
 * File：MagicProtocol.java<br>
 * Title: <br>
 * Description: <br>
 * Company: www.treefinance.com.cn <br>
 * @author heyouchi
 * Date: 2020/5/16
 */
public class MagicProtocol  extends AbstractProtocol {

  public static final String NAME = "magic";

  public static final int DEFAULT_PORT = 20880;
  private static final String IS_CALLBACK_SERVICE_INVOKE = "_isCallBackServiceInvoke";
  private static MagicProtocol INSTANCE;

  private final Map<String, ExchangeServer> serverMap = new ConcurrentHashMap<>();

  private final Map<String, List<ReferenceCountExchangeClient>> referenceClientMap = new ConcurrentHashMap<>();
  private final ConcurrentMap<String, Object> locks = new ConcurrentHashMap<>();
  private final Set<String> optimizers = new ConcurrentHashSet<>();

  private final ConcurrentMap<String, String> stubServiceMethodsMap = new ConcurrentHashMap<>();

  private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {

    @Override
    public CompletableFuture<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {

      if (!(message instanceof Invocation)) {
        throw new RemotingException(channel, "Unsupported request: "
            + (message == null ? null : (message.getClass().getName() + ": " + message))
            + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
      }

      Invocation inv = (Invocation) message;
      Invoker<?> invoker = getInvoker(channel, inv);
      if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
        String methodsStr = invoker.getUrl().getParameters().get("methods");
        boolean hasMethod = false;
        if (methodsStr == null || !methodsStr.contains(",")) {
          hasMethod = inv.getMethodName().equals(methodsStr);
        } else {
          String[] methods = methodsStr.split(",");
          for (String method : methods) {
            if (inv.getMethodName().equals(method)) {
              hasMethod = true;
              break;
            }
          }
        }
        if (!hasMethod) {
          logger.warn(new IllegalStateException("The methodName " + inv.getMethodName()
              + " not found in callback service interface ,invoke will be ignored."
              + " please update the api interface. url is:"
              + invoker.getUrl()) + " ,invocation is :" + inv);
          return null;
        }
      }
      RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
      Result result = invoker.invoke(inv);
      return result.toCompletableFuture().thenApply(Function.identity());
    }

    @Override
    public void received(Channel channel, Object message) throws RemotingException {
      if (message instanceof Invocation) {
        reply((ExchangeChannel) channel, message);

      } else {
        super.received(channel, message);
      }
    }

    @Override
    public void connected(Channel channel) {
      invoke(channel, ON_CONNECT_KEY);
    }

    @Override
    public void disconnected(Channel channel) {
      if (logger.isDebugEnabled()) {
        logger.debug("disconnected from " + channel.getRemoteAddress() + ",url:" + channel.getUrl());
      }
      invoke(channel, ON_DISCONNECT_KEY);
    }

    private void invoke(Channel channel, String methodKey) {
      Invocation invocation = createInvocation(channel, channel.getUrl(), methodKey);
      if (invocation != null) {
        try {
          received(channel, invocation);
        } catch (Throwable t) {
          logger.warn("Failed to invoke event method " + invocation.getMethodName() + "(), cause: " + t.getMessage(), t);
        }
      }
    }

    private Invocation createInvocation(Channel channel, MagicUrl url, String methodKey) {
      String method = url.getParameter(methodKey);
      if (method == null || method.length() == 0) {
        return null;
      }

      RpcInvocation invocation = new RpcInvocation(method, new Class<?>[0], new Object[0]);
      invocation.setAttachment(PATH_KEY, url.getPath());
      invocation.setAttachment(GROUP_KEY, url.getParameter(GROUP_KEY));
      invocation.setAttachment(INTERFACE_KEY, url.getParameter(INTERFACE_KEY));
      invocation.setAttachment(VERSION_KEY, url.getParameter(VERSION_KEY));
      if (url.getParameter(STUB_EVENT_KEY, false)) {
        invocation.setAttachment(STUB_EVENT_KEY, Boolean.TRUE.toString());
      }

      return invocation;
    }
  };

  public MagicProtocol() {
  }

  public static MagicProtocol getMagicProtocol() {
    return (MagicProtocol) MagicFactory.getMagicFactory().getRemoteProtocol(NAME);
  }

  public Collection<ExchangeServer> getServers() {
    return Collections.unmodifiableCollection(serverMap.values());
  }

  public Collection<Exporter<?>> getExporters() {
    return Collections.unmodifiableCollection(exporterMap.values());
  }

  Map<String, Exporter<?>> getExporterMap() {
    return exporterMap;
  }

  private boolean isClientSide(Channel channel) {
    InetSocketAddress address = channel.getRemoteAddress();
    MagicUrl url = channel.getUrl();
    return url.getPort() == address.getPort() &&
        NetUtils.filterLocalHost(channel.getUrl().getIp())
            .equals(NetUtils.filterLocalHost(address.getAddress().getHostAddress()));
  }

  public Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
    boolean isCallBackServiceInvoke = false;
    boolean isStubServiceInvoke = false;
    int port = channel.getLocalAddress().getPort();
    String path = inv.getAttachments().get(PATH_KEY);

    // if it's callback service on client side
    isStubServiceInvoke = Boolean.TRUE.toString().equals(inv.getAttachments().get(STUB_EVENT_KEY));
    if (isStubServiceInvoke) {
      port = channel.getRemoteAddress().getPort();
    }

    //callback
    isCallBackServiceInvoke = isClientSide(channel) && !isStubServiceInvoke;
    if (isCallBackServiceInvoke) {
      path += "." + inv.getAttachments().get(CALLBACK_SERVICE_KEY);
      inv.getAttachments().put(IS_CALLBACK_SERVICE_INVOKE, Boolean.TRUE.toString());
    }
    String serviceKey = serviceKey(port, path, inv.getAttachments().get(VERSION_KEY), inv.getAttachments().get(GROUP_KEY));
    System.out.println("exporterMap.get("+serviceKey+")"+" Attachments"+inv.getAttachments());
    MagicExporter<?> exporter = (MagicExporter<?>) exporterMap.get(serviceKey);

    if (exporter == null) {
      throw new RemotingException(channel, "Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch " +
          ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + inv);
    }

    return exporter.getInvoker();
  }

  public Collection<Invoker<?>> getInvokers() {
    return Collections.unmodifiableCollection(invokers);
  }

  @Override
  public int getDefaultPort() {
    return DEFAULT_PORT;
  }

  @Override
  public <T> Exporter<T> export(Invoker<T> invoker) throws MagicRpcException {
    MagicUrl url = invoker.getUrl();

    // export service.
    String key = serviceKey(url);
    MagicExporter<T> exporter = new MagicExporter<T>(invoker, key, exporterMap);
    System.out.println("exporterMap.put("+key+", "+exporter+")");
    exporterMap.put(key, exporter);

    //export an stub service for dispatching event
    Boolean isStubSupportEvent = url.getParameter(STUB_EVENT_KEY, DEFAULT_STUB_EVENT);
    Boolean isCallbackservice = url.getParameter(IS_CALLBACK_SERVICE, false);
    if (isStubSupportEvent && !isCallbackservice) {
      String stubServiceMethods = url.getParameter(STUB_EVENT_METHODS_KEY);
      if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
        if (logger.isWarnEnabled()) {
          logger.warn(new IllegalStateException("consumer [" + url.getParameter(INTERFACE_KEY) +
              "], has set stubproxy support event ,but no stub methods founded.").getMessage());
        }

      } else {
        stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
      }
    }
    openServer(url);
    return exporter;
  }

  private void openServer(MagicUrl url) {
    // find server.
    String key = url.getAddress();
    //client can export a service which's only for server to invoke
    boolean isServer = url.getParameter(IS_SERVER_KEY, true);
    if (isServer) {
      ExchangeServer server = serverMap.get(key);
      if (server == null) {
        synchronized (this) {
          server = serverMap.get(key);
          if (server == null) {
            serverMap.put(key, createServer(url));
          }
        }
      } else {
        logger.error("key {} ExchangeServer is null", key);
      }
    }
  }

  private ExchangeServer createServer(MagicUrl url) {
    url = MagicUrlBuilder.from(url)
        .addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())
        .addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))
        .build();
    ExchangeServer server;
    try {
      server = Exchangers.bind(url, requestHandler);
    } catch (RemotingException e) {
      throw new MagicRpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
    }
    return server;
  }


  @Override
  public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, MagicUrl url) throws MagicRpcException {
    // create rpc invoker.
    MagicInvoker<T> invoker = new MagicInvoker<T>(serviceType, url, getClients(url), invokers);
    invokers.add(invoker);

    return invoker;
  }

  private ExchangeClient[] getClients(MagicUrl url) {
    // whether to share connection

    boolean useShareConnect = false;

    int connections = url.getParameter(CONNECTIONS_KEY, 0);
    List<ReferenceCountExchangeClient> shareClients = null;
    // if not configured, connection is shared, otherwise, one connection for one service
    if (connections == 0) {
      useShareConnect = true;

      /**
       * The xml configuration should have a higher priority than properties.
       */
      String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
      connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty(SHARE_CONNECTIONS_KEY,
          DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr);
      shareClients = getSharedClient(url, connections);
    }

    ExchangeClient[] clients = new ExchangeClient[connections];
    for (int i = 0; i < clients.length; i++) {
      if (useShareConnect) {
        clients[i] = shareClients.get(i);

      } else {
        clients[i] = initClient(url);
      }
    }

    return clients;
  }

  /**
   * Get shared connection
   *
   * @param url
   * @param connectNum connectNum must be greater than or equal to 1
   */
  private List<ReferenceCountExchangeClient> getSharedClient(MagicUrl url, int connectNum) {
    String key = url.getAddress();
    List<ReferenceCountExchangeClient> clients = referenceClientMap.get(key);

    if (checkClientCanUse(clients)) {
      batchClientRefIncr(clients);
      return clients;
    }

    locks.putIfAbsent(key, new Object());
    synchronized (locks.get(key)) {
      clients = referenceClientMap.get(key);
      // magic check
      if (checkClientCanUse(clients)) {
        batchClientRefIncr(clients);
        return clients;
      }

      // connectNum must be greater than or equal to 1
      connectNum = Math.max(connectNum, 1);

      // If the clients is empty, then the first initialization is
      if (CollectionUtils.isEmpty(clients)) {
        clients = buildReferenceCountExchangeClientList(url, connectNum);
        referenceClientMap.put(key, clients);

      } else {
        for (int i = 0; i < clients.size(); i++) {
          ReferenceCountExchangeClient referenceCountExchangeClient = clients.get(i);
          // If there is a client in the list that is no longer available, create a new one to replace him.
          if (referenceCountExchangeClient == null || referenceCountExchangeClient.isClosed()) {
            clients.set(i, buildReferenceCountExchangeClient(url));
            continue;
          }

          referenceCountExchangeClient.incrementAndGetCount();
        }
      }

      /**
       * I understand that the purpose of the remove operation here is to avoid the expired url key
       * always occupying this memory space.
       */
      locks.remove(key);

      return clients;
    }
  }

  /**
   * Check if the client list is all available
   *
   * @param referenceCountExchangeClients
   * @return true-available，false-unavailable
   */
  private boolean checkClientCanUse(List<ReferenceCountExchangeClient> referenceCountExchangeClients) {
    if (CollectionUtils.isEmpty(referenceCountExchangeClients)) {
      return false;
    }

    for (ReferenceCountExchangeClient referenceCountExchangeClient : referenceCountExchangeClients) {
      // As long as one client is not available, you need to replace the unavailable client with the available one.
      if (referenceCountExchangeClient == null || referenceCountExchangeClient.isClosed()) {
        return false;
      }
    }

    return true;
  }

  /**
   * Increase the reference Count if we create new invoker shares same connection, the connection will be closed without any reference.
   *
   * @param referenceCountExchangeClients
   */
  private void batchClientRefIncr(List<ReferenceCountExchangeClient> referenceCountExchangeClients) {
    if (CollectionUtils.isEmpty(referenceCountExchangeClients)) {
      return;
    }

    for (ReferenceCountExchangeClient referenceCountExchangeClient : referenceCountExchangeClients) {
      if (referenceCountExchangeClient != null) {
        referenceCountExchangeClient.incrementAndGetCount();
      }
    }
  }

  /**
   * Bulk build client
   *
   * @param url
   * @param connectNum
   * @return
   */
  private List<ReferenceCountExchangeClient> buildReferenceCountExchangeClientList(MagicUrl url, int connectNum) {
    List<ReferenceCountExchangeClient> clients = new ArrayList<>();

    for (int i = 0; i < connectNum; i++) {
      clients.add(buildReferenceCountExchangeClient(url));
    }

    return clients;
  }

  /**
   * Build a single client
   *
   * @param url
   * @return
   */
  private ReferenceCountExchangeClient buildReferenceCountExchangeClient(MagicUrl url) {
    ExchangeClient exchangeClient = initClient(url);

    return new ReferenceCountExchangeClient(exchangeClient);
  }

  /**
   * Create new connection
   *
   * @param url
   */
  private ExchangeClient initClient(MagicUrl url) {

    // client type setting.
    //String str = url.getParameter(CLIENT_KEY, url.getParameter(SERVER_KEY, DEFAULT_REMOTING_CLIENT));
    // enable heartbeat by default
    url = url.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT));

    // BIO is not allowed since it has severe performance issue.
    /*if (str != null && str.length() > 0) {
      throw new MagicRpcException("Unsupported client type: " + str + "," +
          " supported client type is " + factory.getTransporter().toString());
    }*/

    ExchangeClient client;
    try {
      // connection should be lazy
      if (url.getParameter(LAZY_CONNECT_KEY, false)) {
        client = new LazyConnectExchangeClient(url, requestHandler);

      } else {
        client = Exchangers.connect(url, requestHandler);
      }

    } catch (RemotingException e) {
      throw new MagicRpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
    }

    return client;
  }

  @Override
  public void destroy() {
    for (String key : new ArrayList<>(serverMap.keySet())) {
      ExchangeServer server = serverMap.remove(key);

      if (server == null) {
        continue;
      }

      try {
        if (logger.isInfoEnabled()) {
          logger.info("Close magic server: " + server.getLocalAddress());
        }

        server.close(CommonConstants.DEFAULT_SERVER_SHUTDOWN_TIMEOUT);

      } catch (Throwable t) {
        logger.warn(t.getMessage(), t);
      }
    }

    for (String key : new ArrayList<>(referenceClientMap.keySet())) {
      List<ReferenceCountExchangeClient> clients = referenceClientMap.remove(key);

      if (CollectionUtils.isEmpty(clients)) {
        continue;
      }

      for (ReferenceCountExchangeClient client : clients) {
        closeReferenceCountExchangeClient(client);
      }
    }

    stubServiceMethodsMap.clear();
    super.destroy();
  }

  /**
   * close ReferenceCountExchangeClient
   *
   * @param client
   */
  private void closeReferenceCountExchangeClient(ReferenceCountExchangeClient client) {
    if (client == null) {
      return;
    }

    try {
      if (logger.isInfoEnabled()) {
        logger.info("Close magic connect: " + client.getLocalAddress() + "-->" + client.getRemoteAddress());
      }

      client.close(CommonConstants.DEFAULT_SERVER_SHUTDOWN_TIMEOUT);

      // TODO
      /**
       * At this time, ReferenceCountExchangeClient#client has been replaced with LazyConnectExchangeClient.
       * Do you need to call client.close again to ensure that LazyConnectExchangeClient is also closed?
       */

    } catch (Throwable t) {
      logger.warn(t.getMessage(), t);
    }
  }
}
